/**
 * Copyright 2005 The Apache Software Foundation
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.mapred;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.UTF8;

/**
 * An {@link InputFormat} for plain text files.  Files are broken into lines.
 * Either linefeed or carriage-return are used to signal end of line.  Keys are
 * the position in the file, and values are the line of text..
 */
public class TextInputFormat extends InputFormatBase {

    @Override
    public RecordReader getRecordReader(FileSystem fs, FileSplit split,
                                        JobConf job, Reporter reporter) throws IOException {

        reporter.setStatus(split.toString());

        final long start = split.getStart();
        final long end = start + split.getLength();

        // open the file and seek to the start of the split
        final FSDataInputStream in = fs.open(split.getFile());

        if (start != 0) {
            in.seek(start - 1);
            // scan to the next newline in the file
            while (in.getPos() < end) {
                char c = (char) in.read();
                if (c == '\r' || c == '\n') {
                    break;
                }
            }
        }

        return new RecordReader() {
            /** Read a line. */
            @Override
            public synchronized boolean next(Writable key, Writable value)
                    throws IOException {
                long pos = in.getPos();
                if (pos >= end) {
                    return false;
                }

                ((LongWritable) key).set(pos);           // key is position
                ((UTF8) value).set(readLine(in));        // value is line
                return true;
            }

            @Override
            public synchronized long getPos() throws IOException {
                return in.getPos();
            }

            @Override
            public synchronized void close() throws IOException {
                in.close();
            }

        };
    }

    /**
     * @author shen
     * @date 2021/10/14
     * 描述：
     * 1、读一行
     */
    private static String readLine(FSDataInputStream in) throws IOException {
        StringBuilder buffer = new StringBuilder();
        while (true) {

            int b = in.read();
            if (b == -1) {
                break;
            }

            // bug: this assumes eight-bit characters.
            char c = (char) b;
            if (c == '\r' || c == '\n') {
                break;
            }

            buffer.append(c);
        }

        return buffer.toString();
    }

}

